use notify::{Config, RecommendedWatcher, Watcher};
use tokio::{
    io::{AsyncBufReadExt, AsyncSeekExt},
    task::JoinHandle,
    time::sleep,
};

use crate::rpc_server::{rpc_server, RpcServerImpl};
use crate::config::AgentConfig;

pub struct MyAgent {
    /// Agent配置信息
    config: AgentConfig,
}

/// new&run
impl MyAgent {
    /// 创建一个新的MyAgent实例
    /// # Arguments
    /// * `config` - Agent配置信息
    /// # Returns
    /// 一个新的MyAgent实例
    pub fn new(config: AgentConfig) -> Self {
        MyAgent { config }
    }

    /// 运行agent
    /// # 核心逻辑
    /// 1. 启动rpc服务器，用于接收来自server的同步文件rpc请求
    /// 2. 向server注册
    /// 3. 监控配置文件变化&&上报异常给server
    pub async fn run(&self) -> anyhow::Result<()> {
        // 启动rpc服务器，用于接收来自server的同步文件rpc请求
        let rpc_handler = self.start_rpc_server().await;

        // 向server注册
        let register_handler = self.register().await;

        // 监控配置文件变化&&上报异常给server
        let monitor_config_file_handler = self.monitor_config_file().await;
        // 等待异步任务完成
        tokio::try_join!(register_handler, rpc_handler, monitor_config_file_handler)?;
        Ok(())
    }
}

// 注册、心跳; 启动rpc服务器
impl MyAgent {
    async fn register(&self) -> JoinHandle<()> {
        // 等待rpc_server启动
        sleep(tokio::time::Duration::from_secs(1)).await;

        let config = self.config.clone();
        // 使用TLS连接
        let tls_config = match crate::tls_config::load_client_tls_config() {
            Ok(config) => config,
            Err(e) => {
                eprintln!("加载TLS配置失败: {:?}", e);
                return tokio::spawn(async {});
            }
        };
        
        let channel = match tonic::transport::Endpoint::from_shared(config.server_addr.clone())
            .unwrap()
            .tls_config(tls_config)
            .unwrap()
            .connect()
            .await {
            Ok(channel) => channel,
            Err(e) => {
                eprintln!("连接Server失败: {:?}", e);
                return tokio::spawn(async {});
            }
        };
        
        let mut client = rpc_server::config_monitor_s_client::ConfigMonitorSClient::new(channel);
        // 向server注册
        tokio::spawn(async move {
            println!("正在向Server注册: {}", config.server_addr);
            println!("Agent ID: {}", config.agent_id);
            println!("Agent标签: {:?}", config.labels);
            
            // 创建AgentInfo
            let agent_info = rpc_server::AgentInfo {
                agent_id: config.agent_id.clone(),
                agent_address: config.get_external_address(),
                hostname: hostname::get().unwrap_or_default().to_string_lossy().to_string(),
                labels: config.labels.clone(),
                last_heartbeat: chrono::Utc::now().timestamp(),
            };
            
            let request = rpc_server::RegisterAgentRequest { 
                agent_info: Some(agent_info)
            };
            
            let response = match client.register_agent(request).await {
                Ok(resp) => resp,
                Err(e) => {
                    eprintln!("注册到Server失败: {:?}", e);
                    return;
                }
            };
            
            let resp = response.into_inner();
            println!("注册成功: {}", resp.message);
            if !resp.assigned_groups.is_empty() {
                println!("自动分配到组: {:?}", resp.assigned_groups);
            }
        })
    }

    async fn start_rpc_server(&self) -> JoinHandle<()> {
        let addr = match self.config.addr.clone().parse() {
            Ok(addr) => addr,
            Err(e) => {
                eprintln!("解析地址失败: {:?}", e);
                return tokio::spawn(async {});
            }
        };

        // 启动rpc服务器
        tokio::spawn(async move {
            println!("启动RPC服务器: {}", addr);
            // 创建rpc服务器
            let rpc_server = RpcServerImpl::new();
            if let Err(e) = tonic::transport::Server::builder()
                .add_service(
                    rpc_server::config_monitor_a_server::ConfigMonitorAServer::new(rpc_server),
                )
                .serve(addr)
                .await
            {
                eprintln!("RPC服务器启动失败: {:?}", e);
            }
        })
    }
}

// 监控配置文件变化，上报异常给server
impl MyAgent {
    const AUDIT_RULE_KEY_PREFIX: &'static str = "key=\"CM_";
    async fn monitor_config_file(&self) -> JoinHandle<()> {
        let server_addr = self.config.server_addr.clone();
        let ip_port = self.config.get_external_address();
        tokio::spawn(async move {
            // 监控audit.log文件，匹配"type=SYSCALL" && "key=\"CM_"，如果匹配到则上报异常

            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

            // notify监控audit.log文件是否有变化
            let mut watcher = match RecommendedWatcher::new(
                move |res| {
                    if let Err(e) = tx.send(res) {
                        eprintln!("发送文件变化事件失败: {:?}", e);
                    }
                },
                Config::default(),
            ) {
                Ok(watcher) => watcher,
                Err(e) => {
                    eprintln!("创建文件监控器失败: {:?}", e);
                    return;
                }
            };
            
            if let Err(e) = watcher.watch(
                std::path::Path::new("/var/log/audit/audit.log"),
                notify::RecursiveMode::Recursive,
            ) {
                eprintln!("监控audit.log文件失败: {:?}", e);
                return;
            }

            let mut file = match tokio::fs::File::open("/var/log/audit/audit.log").await {
                Ok(file) => file,
                Err(e) => {
                    eprintln!("打开audit.log文件失败: {:?}", e);
                    return;
                }
            };
            
            if let Err(e) = file.seek(tokio::io::SeekFrom::End(0)).await {
                eprintln!("定位到文件末尾失败: {:?}", e);
                return;
            }
            let mut reader = tokio::io::BufReader::new(file);
            while let Some(_) = rx.recv().await {
                let mut new_line = String::new();
                loop {
                    match reader.read_line(&mut new_line).await {
                        Ok(0) => break, // EOF
                        Ok(_) => {
                            if new_line.contains("type=SYSCALL")
                                && new_line.contains(Self::AUDIT_RULE_KEY_PREFIX)
                            {
                                // 上报异常
                                #[cfg(debug_assertions)]
                                println!("new_line: {}", new_line);

                                // 上报异常到server
                                match rpc_server::config_monitor_s_client::ConfigMonitorSClient::connect(
                                    server_addr.clone(),
                                ).await {
                                    Ok(mut client) => {
                                        if let Err(e) = client.report_exception(rpc_server::ReportExceptionRequest {
                                            agent_address: ip_port.clone(),
                                            exception_message: new_line.clone(),
                                        }).await {
                                            eprintln!("上报异常失败: {:?}", e);
                                        }
                                    }
                                    Err(e) => {
                                        eprintln!("连接Server失败: {:?}", e);
                                    }
                                }
                            }
                            new_line.clear();
                        }
                        Err(e) => {
                            eprintln!("读取文件行失败: {:?}", e);
                            break;
                        }
                    }
                }
            }
        })
    }
}
